package com.shujia.flink.core

import org.apache.flink.streaming.api.TimerService
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object Demo3Qizha {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    //解析数据
    val transactions: DataStream[Transaction] = linesDS.map(line => {
      val split: Array[String] = line.split(",")
      val accountId: String = split(0)
      val timestamp: String = split(1)
      val amount: Double = split(2).toDouble
      Transaction(accountId, timestamp, amount)
    })

    /**
     * 对于一个账户，如果出现小于 $1 美元的交易后紧跟着一个大于 $500 的交易，就输出一个报警信息。
     * 两个操作必须要在10秒内存完成才发出报警信息
     */
    //1、先安装账号分组
    val keyByDS: KeyedStream[Transaction, String] = transactions.keyBy(_.accountId)


    val alertDS: DataStream[Alert] = keyByDS.process(new KeyedProcessFunction[String, Transaction, Alert] {

      //标记是否出现了小于1美元的数据
      var flag: Boolean = false

      var serviceTime: Long = _

      /**
       * processElement： 每一条数据处理一次
       *
       * @param transaction ：一行数据
       * @param ctx         ：上下文对象
       * @param out         ：用于将数据发送到下游
       */
      override def processElement(transaction: Transaction,
                                  ctx: KeyedProcessFunction[String, Transaction, Alert]#Context,
                                  out: Collector[Alert]): Unit = {
        val amount: Double = transaction.amount
        //小于1美元
        if (amount < 1.0) {
          println(s"$amount,小于1美元")
          flag = true

          //注册一个定时器，10秒之后执行
          val timerService: TimerService = ctx.timerService()
          //当前处理时间
          val currentTime: Long = timerService.currentProcessingTime()
          //定时器执行的时间
          serviceTime = currentTime + 10000
          //注册定时器
          timerService.registerProcessingTimeTimer(serviceTime)
        }

        if (flag) {
          //大于5oo美元
          if (amount > 500) {
            //发送报警信息到下游
            out.collect(Alert(transaction, "操作异常"))

            //清理定时器
            ctx.timerService().deleteProcessingTimeTimer(serviceTime)
            flag = false
          }
        }
      }

      override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, Transaction, Alert]#OnTimerContext, out: Collector[Alert]): Unit = {
        println("定时器执行flag被改成false")
        //已经过去10秒清理掉之前的结果，将flag改成false
        flag = false
      }
    })

    alertDS.print()
    env.execute()
  }

  case class Transaction(accountId: String, timestamp: String, amount: Double)

  case class Alert(transaction: Transaction, msg: String)

}
